<?php
namespace Handlers;

use Com\Alibaba\Otter\Canal\Protocol\Column;
use Com\Alibaba\Otter\Canal\Protocol\Entry;
use Com\Alibaba\Otter\Canal\Protocol\EntryType;
use Com\Alibaba\Otter\Canal\Protocol\EventType;
use Com\Alibaba\Otter\Canal\Protocol\RowChange;
use Com\Alibaba\Otter\Canal\Protocol\RowData;
use Elasticsearch\ClientBuilder as ElasticsearchClient;
use Nette\Database\Connection;

abstract class Handler {
  public $_es_obj;
  public $hosts = [];
  public $log = null;
  public $fields = [];
  public $connect;
  public $instanceConfig;
  public $config;
  
  abstract public function formatData($columns);
  abstract public function insertOrUpdate(array $data, $tableId, $insertRawData);
  abstract public function delete(array $data, $tableId);

  public function __construct($instanceName, $hosts, $log) {
    $this->config = require dirname(__DIR__) . "/configs/config.php";
    $this->instanceConfig = require dirname(__DIR__) . "/configs/{$instanceName}.php";
    $this->hosts = $hosts;
    $this->log = $log;
    $this->connect = $this->getConnect();
  }

  public function syncData($entry) {
    switch ($entry->getEntryType()) {
      case EntryType::TRANSACTIONBEGIN:
      case EntryType::TRANSACTIONEND:
        return;
        break;
    }
    
    $rowChange = new RowChange();
    $rowChange->mergeFromString($entry->getStoreValue());
    $evenType = $rowChange->getEventType();
    $header = $entry->getHeader();

    // echo sprintf(PHP_EOL . "===> binlog[%s : %d],name[%s,%s], eventType: %s", $header->getLogfileName(), $header->getLogfileOffset(), $header->getSchemaName(), $header->getTableName(), $header->getEventType()), PHP_EOL;
    // echo $rowChange->getSql(), PHP_EOL;

    /** @var RowData $rowData */
    foreach ($rowChange->getRowDatas() as $rowData) {
      switch ($evenType) {
        case EventType::INSERT:

          $data = $this->formatData($rowData->getAfterColumns());
          
          if ($this->config['debug'])
            echo "insert event handle:{$data['id']}", PHP_EOL;

          $this->insertOrUpdate($data, 0);

          break;
        case EventType::UPDATE:
          
          $data = $this->formatData($rowData->getAfterColumns());

          if ($this->config['debug'])
            echo "update event handle:{$data['id']}", PHP_EOL;

          $this->insertOrUpdate($data, 0);

          break;
        case EventType::DELETE:

          $data = $this->formatData($rowData->getBeforeColumns());

          if ($this->config['debug'])
            echo "delete event handle:{$data['id']}", PHP_EOL;

          $this->delete($data, 0);

          break;
        default:
          // echo 'others event handle, eventtype:' . $evenType, PHP_EOL;
          break;
      }
    }
  }

  // 获取数据库连接
  private function getConnect() {
    
    $instanceConfig = $this->instanceConfig;

    $dsn = $instanceConfig['database']['dsn'];
    $user = $instanceConfig['database']['user'];
    $password = $instanceConfig['database']['password'];

    $connect = new Connection($dsn, $user, $password);

    return $connect;
  }

  public function changeFieldValueType($name, $value) {
    $fieldType = isset($this->fields[$name]) ? $this->fields[$name] : null;
    
    if ($fieldType === 'json') {
      $value = json_decode($value, true);
    } else if ($fieldType === 'int') {
      $value = (int)$value;
    } else if ($fieldType === 'float') {
      $value = (float)$value;
    } else if ($fieldType === 'datetime') {
      $valueArr = explode(' ', $value);
      $value = $valueArr[0] . 'T' . $valueArr[1];
    } else if ($fieldType === 'boolean') {
      $value = (boolean)$value;
    } else {
      $value = (string)$value;
    }

    return $value;
  }

  public function getEs() {
    if (!$this->_es_obj) {
      try {
        $this->_es_obj = ElasticsearchClient::create()->setHosts($this->hosts)->build();
      } catch (Throwable $e) {
        var_dump(__LINE__, $e->getMessage());
        die;
      }
    }
    return $this->_es_obj;
  }

  public function log($type, $str, $params = []) {
    $this->log && $this->log->$type($str, $params);
  }
   
}